package com.dongql.reactor;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Flow.Publisher;
import java.util.concurrent.Flow.Subscriber;
import java.util.concurrent.Flow.Subscription;
import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.Future;

class OneShotPublisher implements Publisher<Boolean> {

    /**
     * daemon-based
     */
    private final ExecutorService executor = ForkJoinPool.commonPool();

    /**
     * true after first subscribe
     */
    private boolean subscribed;

    @Override
    public synchronized void subscribe(Subscriber<? super Boolean> subscriber) {
        if (subscribed) {
            // only one allowed
            subscriber.onError(new IllegalStateException());
        } else {
            subscribed = true;
            subscriber.onSubscribe(new OneShotSubscription(subscriber, executor));
        }
    }

    static class OneShotSubscription implements Subscription {
        private final Subscriber<? super Boolean> subscriber;
        private final ExecutorService executor;

        /**
         * to allow cancellation
         */
        private Future<?> future;

        private boolean completed;

        OneShotSubscription(Subscriber<? super Boolean> subscriber, ExecutorService executor) {
            this.subscriber = subscriber;
            this.executor = executor;
        }

        @Override
        public synchronized void request(long n) {
            if (n != 0 && !completed) {
                completed = true;
                if (n < 0) {
                    IllegalArgumentException ex = new IllegalArgumentException();
                    executor.execute(() -> subscriber.onError(ex));
                } else {
                    future = executor.submit(() -> {
                        subscriber.onNext(Boolean.TRUE);
                        subscriber.onComplete();
                    });
                }
            }
        }

        @Override
        public synchronized void cancel() {
            completed = true;
            if (future != null) {
                future.cancel(false);
            }
        }
    }
}
